package com.etc.sparkstrea

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * 基于transform的实时广告计费日志黑名单过滤
  * 这里案例，完全脱胎于真实的广告业务的大数据系统，业务是真实的，实用
  * @author Administrator
  *
  */
object TransformBlacklist {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("TransformBlacklist").setMaster("local[*]")
    val sc = new StreamingContext(conf,Seconds(5))
    val blacklist = Array(("tom",true));
    val line = sc.sparkContext.parallelize(blacklist)
    val line1 = sc.socketTextStream("master",8888)
    val unit = line1.map { a => (a.split(" ")(1), a) }.transform(a => {
      val value = a.leftOuterJoin(line)
      value.filter(t => {
        if (t._2._2.getOrElse(false)) {
          false
        } else {
          true
        }
      }).map(t => t._2._1)
      value
    })
    unit.print()
    sc.start()
    sc.awaitTermination()

  }

}
